/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.scheduler;

import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import cgl.granules.dataset.DatasetException;
import cgl.granules.operation.Operation;
import cgl.granules.operation.ProcessingException;
import cgl.narada.util.UUIDRetriever;

/**
 * This is the class that is responsible for managing tasks.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class TaskManagement implements RejectedExecutionHandler,
        TaskLifecycleObserver {
    private int numberOfThreads;

    private int maxQueueCapacity;

    private ScheduledThreadPoolExecutor executor;

    private ConcurrentHashMap<String, ExecutableTask> pendingTasks;
    private ConcurrentHashMap<String, String> waitingTasks;
    private ConcurrentHashMap<String, ScheduledFuture<?>> futureTasks;

    private ConcurrentLinkedQueue<String> executionOrder;
    private UUIDRetriever retriever;

    private Diagnostics diagnostics;
    private boolean debug = false;


    public TaskManagement(int numberOfThreads, Diagnostics diagnostics) {

        this.numberOfThreads = numberOfThreads;
        this.diagnostics = diagnostics;

        executor = new ScheduledThreadPoolExecutor(numberOfThreads, this);
        executor.prestartAllCoreThreads();
        pendingTasks = new ConcurrentHashMap<String, ExecutableTask>();
        waitingTasks = new ConcurrentHashMap<String, String>();
        futureTasks = new ConcurrentHashMap<String, ScheduledFuture<?>>();
        executionOrder = new ConcurrentLinkedQueue<String>();
        retriever = UUIDRetriever.getInstance();
    }


    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        ExecutableTask rejectedTask = (ExecutableTask) r;
        System.out.println("The execution of task ( " + rejectedTask
            + " was rejected!");
        String taskIdentifier = rejectedTask.getIdentifier();
        rejectedTask.setExecutionStatus(TaskLifecycleMetrics.ABORTED);
        cleanUpAndReportOnTask(taskIdentifier);

        /* Issue an error notification to the originator. */
    }


    @Override
    public void abnormalOperationTermination(String taskIdentifier,
                                             String reason) {
        ExecutableTask abortedTask = pendingTasks.get(taskIdentifier);
        abortedTask.setExecutionStatus(TaskLifecycleMetrics.ABORTED);
        cleanUpAndReportOnTask(taskIdentifier);
        System.out.println("Task (" + taskIdentifier
            + ") terminated abnormally because: " + reason);
    }


    /**
     * Manage the execution of the given operation
     * 
     * @param taskIdentifier
     *            The identifier of the task being started up.
     * @param operation
     *            The operation that needs to be submitted for execution
     * @throws ProcessingException
     *             If there are problems initializing the operation
     * @throws DatasetException
     *             If there are problems in accessing the underling dataset
     * @throws SchedulingException
     *             If there are problems in scheduling the operation
     */
    public void manageExecution(String taskIdentifier, Operation operation)
        throws ProcessingException, DatasetException, SchedulingException {

        ExecutableTask executableTask =
            new ExecutableTask(taskIdentifier, operation, this);

        pendingTasks.put(taskIdentifier, executableTask);
        diagnostics.createdTask(taskIdentifier,
                                executableTask.getTaskLifecycleMetrics());

        if (executableTask.passesReadinessCheck()) {
            if (debug) {
                System.out.println("Task " + taskIdentifier
                    + " passes readiness check");
            }
            waitingTasks.put(taskIdentifier, taskIdentifier);
            executionOrder.add(taskIdentifier);
            tryAndSubmitTasks();
        }

    }


    /**
     * Performs cleanup and diagnostic reporting once a task is deemed to
     * finished execution, either successfully or because of some errors. This
     * information is encapsulated in the LifecyleMetrics associated with the
     * task. Additionally, an attempt is made to schedule additional tasks since
     * the execution queue has now been reduced by 1.
     * 
     * @param taskIdentifier
     *            The identifier of the task being cleaned up.
     */
    private void cleanUpAndReportOnTask(String taskIdentifier) {
        /* Remove the task from the list of executable tasks */
        ExecutableTask task = pendingTasks.remove(taskIdentifier);
        waitingTasks.remove(taskIdentifier);
        if (debug) {
            System.out.println("Trying to remove: " + task.getIdentifier());
        }

        diagnostics.completedTask(taskIdentifier);
        if (task.isExecutePeriodically()) {
            ScheduledFuture<?> future = futureTasks.get(taskIdentifier);
            if (future != null) {
                future.cancel(true);
                executor.remove(task);
                executor.purge();
            }
        }

        // task = null;

        tryAndSubmitTasks();
    }


    @Override
    public void isComplete(String taskIdentifier) {
        cleanUpAndReportOnTask(taskIdentifier);

        if (debug) {
            String report =
                "Task (" + taskIdentifier + ") is complete. "
                    + "Execution Queue Size = " + executor.getQueue().size()
                    + " Pending Tasks =" + pendingTasks.size()
                    + " Executing Order Size =" + executionOrder.size()
                    + " Waiting Tasks = " + waitingTasks.size();
            System.out.println(report);
        }
        // issueCompletionNotification(taskIdentifier);
    }


    @Override
    public void isExecuting(String taskIdentifier) {
        waitingTasks.remove(taskIdentifier);
    }


    @Override
    public void isInitializing(String taskIdentifier) {
        if (debug) {
            System.out.println("Task (" + taskIdentifier + ") is ready!");
        }
    }


    @Override
    public void isReady(String taskIdentifier) {
        if (debug) {
            System.out.println("Task (" + taskIdentifier + ") is ready!");
        }
        if (waitingTasks.containsKey(taskIdentifier)) {
            return;
        }
        waitingTasks.put(taskIdentifier, taskIdentifier);
        executionOrder.add(taskIdentifier);
        tryAndSubmitTasks();
    }


    @Override
    public void isWaitingForData(String taskIdentifier) {
        if (debug) {
            System.out.println("Task (" + taskIdentifier
                + ") waiting for data!");
        }
    }


    @Override
    public void isChangedPeriodicExecution(String taskIdentifier) {
        ExecutableTask task = pendingTasks.get(taskIdentifier);
        if (task == null) {
            return;
        }

        /** This assumes that task was previously executed periodically. */
        ScheduledFuture<?> future = futureTasks.get(taskIdentifier);
        if (future != null) {
            future.cancel(true);
            executor.remove(task);
            executor.purge();
        }
        executionOrder.add(taskIdentifier);
        tryAndSubmitTasks();
    }


    /**
     * Check the various data structures and determine if additional tasks can
     * be scheduled for execution.
     */
    private void tryAndSubmitTasks() {
        try {
            if (executionOrder.isEmpty()) {
                if (debug) {
                    System.out.println("Execution Queue is empty!");
                }
            }

            while (!executionOrder.isEmpty()) {
                // if (executor.getQueue().size() >= maxQueueCapacity - 10) {
                // System.out.println("Maximum capacity limit reached!");
                // break;
                // }
                String taskId = executionOrder.poll();
                ExecutableTask task = pendingTasks.get(taskId);

                if (task == null) {
                    System.out.println("Task " + taskId
                        + " no longer present in the set of manageable tasks."
                        + " Most likely task was completed, and reclaimed!");
                    continue;
                }

                if (task.isExecutePeriodically()) {
                    long executionInterval = task.getExecutionInterval();
                    long initialDelay = task.getInitialDelayForPeriodicTask();

                    ScheduledFuture<?> future =
                        executor.scheduleWithFixedDelay(task, initialDelay,
                                                        executionInterval,
                                                        TimeUnit.MILLISECONDS);
                    futureTasks.put(taskId, future);
                    if (debug) {
                        System.out.println("Submitted task [" + taskId
                            + "] for periodic execution!");
                    }
                } else {
                    executor.execute(task);
                    if (debug) {
                        System.out.println("Submitted task " + taskId
                            + " for execution");
                    }
                }
            }
        } catch (NoSuchElementException nsee) {
            System.out.println(nsee);
            return;
        }
    }

}
